package com.shu.bolt;

import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

public class PrintOutput extends BaseRichBolt{
	
	private OutputCollector collector;

	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this.collector = collector;
	}

	public void execute(Tuple input) {
		// 从上面获取分词的累计次数
		try {
			String name = input.getStringByField("word");
		    Integer counts = input.getIntegerByField("count");
		    System.out.println("==========================================="+name+":"+counts);
		    collector.ack(input);
		}catch(Exception e) {
			collector.fail(input);
		}
      
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		
	}

}
